* Thu Sep 05 2013 - Add more diagrams to 'Overview' * Wed Sep 04 2013 - Finish first chapter 'Overview'
In Progress
Overview
JGroup is the corner stone for clustering. It is a reliable multicast system. As we know, using IP multicasting we can set a group of computers to use one multicast IP address, and all the messages sent to this multicast IP address will be received by all the computers in the group.
IP multicasting is usually used together with UDP protocol. Because UDP protocol is stateless, if one computer send an UDP datagram to one multicast IP address, this UDP datagram will be sent to all the members in the group. Meanwhile, the sender of this UDP datagram don't care whether destination computers received this datagram or not. Here is the diagram that shows a IP multicasting group that contains three members:
Here is the use case for above scenario:
UDP is suitable for this scenario. Because if we use TCP, then the sender must establish connections with all the members in a multicasting group, and check the messages are received successfully by all the members.
However, UDP is not a reliable protocol, the datagram maybe lost during transmission and it won't be checked:
Besides, UDP datagrams may also arrived in random sequence and not sorted by timeline: If the sender sends packet 1 firstly, and then packet 2, the receiver may receive packet 2 firstly and then 1:
So if we want to build a reliable cluster system based on IP multicast + UDP, we need to ensure the data stability by ourselves.
One example is like numbering all the datagrams we've sent in application layer. And the receiver will need to check the data is not lost and then sort the packets into correct order. For example:
-
If sender sent packet 1, 2 to multicast group, and in this group there are two receivers called A and B. ReceiverA received packet 1, 2 successfully, but receiver B only get packet 1. Then receiverB will ask sender to retransmit packet2. Please note this time the sender don't have to send packet 1 to multicast address, because other receivers in the group didn't ask for it because they have gotten the packet 2 successfully. Sender just need to send packet 1 to receiverB again with an unicast to reduce bandwidth cost:
-
If sender sent packet 1, packet 2, and receiverA received packet 2, packet 1, it will reorder the packet to packet 1, packet 2 by itself.
In above is some simple design concepts on how to build a reliable transmission system on top of IP multicasting + UDP. JGroup is certainly much more complex than above example but the design concept is basically the same.
In JGroup, the minimal group unit is called 'Channel'. Members in a channel can talk with each other. The data send/receive in a channel is called 'Message'. Here is the use case:
It's very similar to IP multicasting + UDP:
Generally speaking, JGroup is designed as a 'layered protocol stack'. In the bottom of the stack it's the 'UDP' protocol. Other protocols are built on top of it to meet different requirements of a cluster. For example, some protocol ensures the packet reliability (like the example shown above: all the packets should be sent to the members in cluster with 0% data loss); some protocol will check the liveness of the members in a cluster (for example, ping the members periodically); and some protocol will cut the big datagram into smaller one for UDP to be able to transmit them (as we know, the UDP packet is sensitive to the size of datagram).
Here is the default setup of JGroup protocol stack:
You can even guess some protocol's purpose by its name. The complete description of each protocol could be found here:
I wish you have caught the general design concepts of JGroup with above description. Now let's get into the code:
Developing a sample project
In JGroup a cluster is called a channel. The members in the channel can send messages to or receive messages from this channel. We will develop an example that creates a channel and have two members in it: A sender that sends messages to the channel and a receiver that receive messages from the channel.
Please note a member in a channel can be both a sender a receiver. We just simplify the scenario in the example for the code to be better explained. Here is the use case:
From the above diagram we can see the first step to setup a cluster is to ask members to join to a same channel, then they can communicate with other members in this group. In below are the real codes:
import org.jgroups.JChannel;
import org.jgroups.Message;
import java.io.BufferedReader;
import java.io.InputStreamReader;
public class ChatSender {
private JChannel channel;
private void start() throws Exception {
channel = new JChannel();
channel.connect("ChatCluster");
eventLoop();
channel.close();
}
private void eventLoop() {
BufferedReader in = new BufferedReader(new InputStreamReader(System.in));
while (true) {
try {
System.out.print("> ");
System.out.flush();
String line = in.readLine().toLowerCase();
if (line.startsWith("quit") || line.startsWith("exit")) {
break;
}
Message msg = new Message(null, null, line);
channel.send(msg);
} catch (Exception e) {
}
}
}
public static void main(String[] args) throws Exception {
ChatSender play = new ChatSender();
play.start();
}
}
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.ReceiverAdapter;
import org.jgroups.View;
public class ChatReceiver extends ReceiverAdapter {
public void viewAccepted(View new_view) {
System.out.println("** view: " + new_view);
}
public void receive(Message msg) {
System.out.println(msg.getSrc() + ": " + msg.getObject());
}
public void start() throws Exception {
JChannel channel = new JChannel();
channel.setReceiver(this);
channel.connect("ChatCluster");
}
public static void main(String[] args) throws Exception {
new ChatReceiver().start();
}
}
Workflow
Channel Initialization
The first step is to create a JChannel instance:
JChannel channel = new JChannel();
Load default protocol stack
If no setting provided, JChannel will provide a default protocol stack:
public JChannel() throws Exception {
this(DEFAULT_PROTOCOL_STACK);
}
The default stack is like this:
public static final String DEFAULT_PROTOCOL_STACK="udp.xml";
<!--
Default stack using IP multicasting. It is similar to the "udp"
stack in stacks.xml, but doesn't use streaming state transfer and flushing
author: Bela Ban
-->
<config xmlns="urn:org:jgroups"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="urn:org:jgroups http://www.jgroups.org/schema/JGroups-3.3.xsd">
<UDP
mcast_port="${jgroups.udp.mcast_port:45588}"
tos="8"
ucast_recv_buf_size="5M"
ucast_send_buf_size="640K"
mcast_recv_buf_size="5M"
mcast_send_buf_size="640K"
loopback="true"
max_bundle_size="64K"
max_bundle_timeout="30"
ip_ttl="${jgroups.udp.ip_ttl:8}"
enable_diagnostics="true"
thread_naming_pattern="cl"
timer_type="new3"
timer.min_threads="4"
timer.max_threads="10"
timer.keep_alive_time="3000"
timer.queue_max_size="500"
thread_pool.enabled="true"
thread_pool.min_threads="2"
thread_pool.max_threads="8"
thread_pool.keep_alive_time="5000"
thread_pool.queue_enabled="true"
thread_pool.queue_max_size="10000"
thread_pool.rejection_policy="discard"
oob_thread_pool.enabled="true"
oob_thread_pool.min_threads="1"
oob_thread_pool.max_threads="8"
oob_thread_pool.keep_alive_time="5000"
oob_thread_pool.queue_enabled="false"
oob_thread_pool.queue_max_size="100"
oob_thread_pool.rejection_policy="discard"/>
<PING timeout="2000"
num_initial_members="20"/>
<MERGE2 max_interval="30000"
min_interval="10000"/>
<FD_SOCK/>
<FD_ALL/>
<VERIFY_SUSPECT timeout="1500" />
<BARRIER />
<pbcast.NAKACK2 xmit_interval="500"
xmit_table_num_rows="100"
xmit_table_msgs_per_row="2000"
xmit_table_max_compaction_time="30000"
max_msg_batch_size="500"
use_mcast_xmit="false"
discard_delivered_msgs="true"/>
<UNICAST3 xmit_interval="500"
xmit_table_num_rows="100"
xmit_table_msgs_per_row="2000"
xmit_table_max_compaction_time="60000"
conn_expiry_timeout="0"
max_msg_batch_size="500"/>
<pbcast.STABLE stability_delay="1000" desired_avg_gossip="50000"
max_bytes="4M"/>
<pbcast.GMS print_local_addr="true" join_timeout="3000"
view_bundling="true"/>
<UFC max_credits="2M"
min_threshold="0.4"/>
<MFC max_credits="2M"
min_threshold="0.4"/>
<FRAG2 frag_size="60K" />
<RSVP resend_interval="2000" timeout="10000"/>
<pbcast.STATE_TRANSFER />
<!-- pbcast.FLUSH /-->
</config>
Each protocol is focusing on its own task, for a complete definition of each stack, please check JGroup Protocol Link
After the loading process, the xml file is loaded into ProtocolStackConfigurator:
/**
* Constructs a JChannel instance with the protocol stack configuration based upon the specified properties parameter.
* @param props A file containing a JGroups XML configuration, a URL pointing to an XML configuration, or an old
* style plain configuration string.
* @throws Exception If problems occur during the configuration or initialization of the protocol stack.
*/
public JChannel(String props) throws Exception {
this(ConfiguratorFactory.getStackConfigurator(props));
}
Channel initialization
After loading the configure, JChannel will go on its initialization process:
/**
* Constructs a JChannel with the protocol stack configuration contained by the protocol stack configurator parameter.
* <p>
* All of the public constructors of this class eventually delegate to this method.
* @param configurator A protocol stack configurator containing a JGroups protocol stack configuration.
* @throws Exception If problems occur during the initialization of the protocol stack.
*/
public JChannel(ProtocolStackConfigurator configurator) throws Exception {
init(configurator);
}
protected final void init(ProtocolStackConfigurator configurator) throws Exception {
prot_stack=new ProtocolStack(this);
prot_stack.setup(configs); // Setup protocol stack (creates protocol, calls init() on them)
}
prot_stack=new ProtocolStack(this);
A ProtocolStack instance is created with the current Channel config, and two xml files are loaded into ClassConfigurator:
protected static void init() throws Exception {
// Read jg-magic-map.xml
List<Tuple<Short,String>> mapping=readMappings(magic_number_file);
mapping=readMappings(protocol_id_file); // Read jg-protocol-ids.xml
}
jg-magic-map.xml
<?xml version="1.0" encoding="UTF-8"?>
<magic-number-class-mapping>
<class id="1" name="org.jgroups.stack.IpAddress"/>
<class id="3" name="org.jgroups.protocols.FD$FdHeader"/>
<class id="6" name="org.jgroups.protocols.FD_SOCK$FdHeader"/>
<class id="7" name="org.jgroups.protocols.FragHeader"/>
<class id="13" name="org.jgroups.protocols.PingHeader"/>
<class id="21" name="org.jgroups.protocols.UNICAST$UnicastHeader"/>
<class id="22" name="org.jgroups.protocols.VERIFY_SUSPECT$VerifyHeader"/>
<class id="24" name="org.jgroups.protocols.pbcast.GMS$GmsHeader"/>
<class id="25" name="org.jgroups.protocols.pbcast.NakAckHeader"/>
<class id="27" name="org.jgroups.protocols.pbcast.STABLE$StableHeader"/>
<class id="28" name="org.jgroups.protocols.pbcast.STATE_TRANSFER$StateHeader"/>
<class id="30" name="org.jgroups.Message"/>
<class id="31" name="org.jgroups.View"/>
<class id="32" name="org.jgroups.ViewId"/>
<class id="34" name="org.jgroups.Address"/>
<class id="36" name="org.jgroups.protocols.PingData"/>
<class id="38" name="java.util.Vector"/>
<class id="39" name="org.jgroups.protocols.pbcast.JoinRsp"/>
<class id="40" name="org.jgroups.util.Digest"/>
<class id="41" name="java.util.Hashtable"/>
<class id="53" name="org.jgroups.protocols.COMPRESS$CompressHeader"/>
<class id="54" name="org.jgroups.protocols.FcHeader"/>
<class id="56" name="org.jgroups.protocols.TpHeader"/>
<class id="57" name="org.jgroups.protocols.ENCRYPT$EncryptHeader"/>
<class id="58" name="org.jgroups.protocols.SEQUENCER$SequencerHeader"/>
<class id="61" name="org.jgroups.protocols.FD_ALL$HeartbeatHeader"/>
<class id="64" name="org.jgroups.protocols.pbcast.FLUSH$FlushHeader"/>
<class id="65" name="org.jgroups.protocols.pbcast.StreamingStateTransfer$StateHeader"/>
<class id="67" name="org.jgroups.protocols.AuthHeader"/>
<class id="68" name="org.jgroups.util.UUID"/>
<class id="71" name="org.jgroups.blocks.RequestCorrelator$Header"/>
<class id="72" name="org.jgroups.blocks.RequestCorrelator$MultiDestinationHeader"/>
<class id="73" name="org.jgroups.protocols.UNICAST2$Unicast2Header"/>
<class id="74" name="org.jgroups.protocols.SCOPE$ScopeHeader"/>
<class id="75" name="org.jgroups.blocks.mux.MuxHeader"/>
<class id="76" name="org.jgroups.protocols.DAISYCHAIN$DaisyHeader"/>
<class id="77" name="org.jgroups.protocols.RELAY$RelayHeader"/>
<class id="78" name="org.jgroups.protocols.STOMP$StompHeader"/>
<class id="80" name="org.jgroups.protocols.PrioHeader"/>
<class id="81" name="org.jgroups.protocols.Locking$LockingHeader"/>
<class id="82" name="org.jgroups.util.PayloadUUID"/>
<class id="83" name="org.jgroups.util.AdditionalDataUUID"/>
<class id="84" name="org.jgroups.util.TopologyUUID"/>
<class id="85" name="org.jgroups.protocols.Executing$ExecutorHeader"/>
<class id="86" name="org.jgroups.protocols.Executing$Request"/>
<class id="87" name="org.jgroups.blocks.executor.ExecutionService$RunnableAdapter"/>
<class id="88" name="org.jgroups.blocks.executor.Executions$StreamableCallable"/>
<class id="89" name="org.jgroups.protocols.COUNTER$CounterHeader"/>
<class id="90" name="org.jgroups.protocols.MERGE3$MergeHeader"/>
<class id="91" name="org.jgroups.protocols.RSVP$RsvpHeader"/>
<class id="92" name="org.jgroups.tests.perf.MPerf$MPerfHeader"/>
<class id="93" name="org.jgroups.protocols.pbcast.NakAckHeader2"/>
<class id="94" name="org.jgroups.util.SeqnoList"/>
<class id="95" name="org.jgroups.protocols.tom.ToaHeader"/>
<class id="96" name="org.jgroups.AnycastAddress"/>
<class id="97" name="org.jgroups.protocols.relay.SiteUUID"/>
<class id="98" name="org.jgroups.protocols.relay.SiteMaster"/>
<class id="99" name="org.jgroups.protocols.relay.RELAY2$Relay2Header"/>
<class id="100" name="org.jgroups.protocols.FORWARD_TO_COORD$ForwardHeader"/>
<class id="101" name="org.jgroups.protocols.relay.CanBeSiteMaster"/>
<class id="102" name="org.jgroups.protocols.relay.CanBeSiteMasterTopology"/>
<class id="103" name="org.jgroups.auth.SimpleToken"/>
<class id="104" name="org.jgroups.auth.FixedMembershipToken"/>
<class id="105" name="org.jgroups.auth.MD5Token"/>
<class id="106" name="org.jgroups.auth.X509Token"/>
<class id="107" name="org.jgroups.protocols.UNICAST3$Header"/>
</magic-number-class-mapping>
jg-protocol-ids.xml
<?xml version="1.0" encoding="UTF-8"?>
<protocol-ids>
<!-- Reserved IDs -->
<class id="2" name="org.jgroups.protocols.FD"/>
<class id="3" name="org.jgroups.protocols.FD_SOCK"/>
<class id="4" name="org.jgroups.protocols.FRAG"/>
<class id="5" name="org.jgroups.protocols.FRAG2"/>
<class id="6" name="org.jgroups.protocols.PING"/>
<class id="7" name="org.jgroups.protocols.MPING"/>
<class id="8" name="org.jgroups.protocols.S3_PING"/>
<class id="9" name="org.jgroups.protocols.FILE_PING"/>
<class id="10" name="org.jgroups.protocols.TCPPING"/>
<class id="11" name="org.jgroups.protocols.TCPGOSSIP"/>
<class id="12" name="org.jgroups.protocols.UNICAST"/>
<class id="13" name="org.jgroups.protocols.VERIFY_SUSPECT"/>
<class id="14" name="org.jgroups.protocols.pbcast.GMS"/>
<class id="15" name="org.jgroups.protocols.pbcast.NAKACK"/>
<class id="16" name="org.jgroups.protocols.pbcast.STABLE"/>
<class id="17" name="org.jgroups.protocols.pbcast.STATE_TRANSFER"/>
<class id="19" name="org.jgroups.protocols.COMPRESS"/>
<class id="20" name="org.jgroups.protocols.FC"/>
<class id="21" name="org.jgroups.protocols.UDP"/>
<class id="22" name="org.jgroups.protocols.TCP"/>
<class id="23" name="org.jgroups.protocols.TCP_NIO"/>
<class id="24" name="org.jgroups.protocols.TUNNEL"/>
<class id="25" name="org.jgroups.protocols.ENCRYPT"/>
<class id="26" name="org.jgroups.protocols.SEQUENCER"/>
<class id="29" name="org.jgroups.protocols.FD_ALL"/>
<class id="31" name="org.jgroups.protocols.pbcast.FLUSH"/>
<class id="33" name="org.jgroups.protocols.AUTH"/>
<class id="34" name="org.jgroups.protocols.pbcast.STATE"/>
<class id="35" name="org.jgroups.protocols.pbcast.STATE_SOCK"/>
<class id="37" name="org.jgroups.protocols.DISCARD"/>
<class id="39" name="org.jgroups.protocols.SHARED_LOOPBACK"/>
<class id="40" name="org.jgroups.protocols.UNICAST2"/>
<class id="41" name="org.jgroups.protocols.SCOPE"/>
<class id="42" name="org.jgroups.protocols.DAISYCHAIN"/>
<class id="43" name="org.jgroups.protocols.RELAY"/>
<class id="44" name="org.jgroups.protocols.MFC"/>
<class id="45" name="org.jgroups.protocols.UFC"/>
<class id="46" name="org.jgroups.protocols.JDBC_PING"/>
<class id="47" name="org.jgroups.protocols.STOMP"/>
<class id="48" name="org.jgroups.protocols.PRIO"/>
<class id="49" name="org.jgroups.protocols.BPING"/>
<class id="50" name="org.jgroups.protocols.CENTRAL_LOCK"/>
<class id="51" name="org.jgroups.protocols.PEER_LOCK"/>
<class id="52" name="org.jgroups.protocols.CENTRAL_EXECUTOR"/>
<class id="53" name="org.jgroups.protocols.COUNTER"/>
<class id="54" name="org.jgroups.protocols.MERGE3"/>
<class id="55" name="org.jgroups.protocols.RSVP"/>
<class id="56" name="org.jgroups.protocols.RACKSPACE_PING"/>
<class id="57" name="org.jgroups.protocols.pbcast.NAKACK2"/>
<class id="58" name="org.jgroups.protocols.tom.TOA"/>
<class id="59" name="org.jgroups.protocols.SWIFT_PING"/>
<class id="60" name="org.jgroups.protocols.relay.RELAY2"/>
<class id="61" name="org.jgroups.protocols.FORWARD_TO_COORD"/>
<class id="62" name="org.jgroups.protocols.PDC"/>
<class id="63" name="org.jgroups.protocols.rules.SUPERVISOR"/>
<class id="64" name="org.jgroups.protocols.UNICAST3"/>
<!-- IDs reserved for building blocks -->
<class id="200" name="org.jgroups.blocks.RequestCorrelator"/> <!-- ID should be the same as Global.BLOCKS_START_ID -->
<class id="201" name="org.jgroups.blocks.mux.MuxRequestCorrelator"/>
<class id="202" name="org.jgroups.tests.perf.MPerf"/>
<!-- User-defined protocols should use IDs >= 1000 -->
</protocol-ids>
prot_stack.setup(configs);
Setup protocol stack (creates protocol, calls init() on them)
public void setup(List<ProtocolConfiguration> configs) throws Exception {
if(top_prot == null) {
top_prot=new Configurator(this).setupProtocolStack(configs);
top_prot.setUpProtocol(this);
this.setDownProtocol(top_prot);
bottom_prot=getBottomProtocol();
initProtocolStack();
}
}
In the situation of default configuration:
-
top_prot == STATE_TRANSFER
-
bottom_prot == UDP
-
initProtocolStack will setup each protocol in the stack:
public void initProtocolStack() throws Exception {
List<Protocol> protocols = getProtocols();
Collections.reverse(protocols);
for(Protocol prot: protocols) {
...
prot.init();
}
}
Connect to channel
channel.connect("ChatCluster");
@ManagedOperation(description="Connects the channel to a group")
public synchronized void connect(String cluster_name) throws Exception {
connect(cluster_name, true);
}
@ManagedOperation(description="Connects the channel to a group")
protected synchronized void connect(String cluster_name, boolean useFlushIfPresent) throws Exception {
if(!_preConnect(cluster_name))
return;
if(cluster_name != null) { // only connect if we are not a unicast channel
Event connect_event=useFlushIfPresent? new Event(Event.CONNECT_USE_FLUSH, cluster_name)
: new Event(Event.CONNECT, cluster_name);
_connect(connect_event);
}
state=State.CONNECTED;
notifyChannelConnected(this);
}
_preConnect(cluster_name); _connect(connect_event); state=State.CONNECTED; notifyChannelConnected(this);
_preConnect
protected boolean _preConnect(String cluster_name) throws Exception {
if(state == State.CONNECTED) {
if(log.isTraceEnabled()) log.trace("already connected to " + this.cluster_name);
return false;
}
checkClosed();
setAddress();
State old_state=state;
state=State.CONNECTING;
try {
startStack(cluster_name);
}
catch(Exception ex) {
state=old_state;
throw ex;
}
return true;
}
setAddress
protected void setAddress() {
Address old_addr=local_addr;
local_addr=address_generator != null? address_generator.generateAddress() : UUID.randomUUID();
if(old_addr != null)
down(new Event(Event.REMOVE_ADDRESS, old_addr));
if(name == null || name.isEmpty()) // generate a logical name if not set
name=Util.generateLocalName();
if(name != null && !name.isEmpty())
UUID.add(local_addr, name);
Event evt=new Event(Event.SET_LOCAL_ADDRESS, local_addr);
down(evt);
if(up_handler != null)
up_handler.up(evt);
}
local_addr=address_generator != null? address_generator.generateAddress() : UUID.randomUUID(); name=Util.generateLocalName();
down(evt);
/**
* Sends an event down the protocol stack. Note that - contrary to {@link #send(Message)}, if the event is a message,
* no checks are performed whether the channel is closed or disconnected.
* @param evt the message to send down, encapsulated in an event
*/
public Object down(Event evt) {
if(evt == null) return null;
return prot_stack.down(evt);
}
startStack
protected void startStack(String cluster_name) throws Exception {
/*make sure the channel is not closed*/
checkClosed();
/*make sure we have a valid channel name*/
if(cluster_name == null) {
if(log.isDebugEnabled()) log.debug("cluster_name is null, assuming unicast channel");
}
else
this.cluster_name=cluster_name;
if(socket_factory != null)
prot_stack.getTopProtocol().setSocketFactory(socket_factory);
prot_stack.startStack(cluster_name, local_addr); // calls start() in all protocols, from top to bottom
/*create a temporary view, assume this channel is the only member and is the coordinator*/
List<Address> t=new ArrayList<Address>(1);
t.add(local_addr);
my_view=new View(local_addr, 0, t); // create a dummy view
TP transport=prot_stack.getTransport();
transport.registerProbeHandler(probe_handler);
}
prot_stack.getTopProtocol().setSocketFactory(socket_factory);
In Protocol.java:
public void setSocketFactory(SocketFactory factory) {
if(down_prot != null)
down_prot.setSocketFactory(factory);
}
prot_stack.startStack(cluster_name, local_addr);
TP transport=prot_stack.getTransport();
transport.registerProbeHandler(probe_handler);